package net.bwie.flink;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.TableEnvironment;

/**
 * 基于FlinkSQL，直接消费Kafka队列中order_detail订单明细数据，采用SQL实时统计
 * @author xuanyu
 * @date 2025/10/28
 */
public class _08SqlOrderDemo {

	public static void main(String[] args) {
		// 1.表执行环境-tabEnv
		EnvironmentSettings settings = EnvironmentSettings
			.newInstance()
			.inStreamingMode()
			.build();
		TableEnvironment tabEnv = TableEnvironment.create(settings);
		// 设置属性，比如并行度为1 、
		/*
		https://nightlies.apache.org/flink/flink-docs-release-1.17/docs/dev/table/config/#execution-options
		 */
		TableConfig tableConfig = tabEnv.getConfig();
		tableConfig.set("table.exec.resource.default-parallelism", "1");

		// 2.输入表-inputTable
		tabEnv.executeSql(
			"CREATE TABLE mysql_cdc_kafka_source (\n" +
				"    `before` MAP<STRING, STRING>,\n" +
				"    `after` MAP<STRING, STRING>,\n" +
				"    `source` MAP<STRING, STRING>,\n" +
				"    `op` STRING,\n" +
				"    `ts_ms` BIGINT,\n" +
				"    `transaction` STRING\n" +
				") WITH (\n" +
				"  'connector' = 'kafka',\n" +
				"  'topic' = 'order_info',\n" +
				"  'properties.bootstrap.servers' = 'node101:9092,node102:9092,node103:9092',\n" +
				"  'properties.group.id' = 'order_info_g1',\n" +
				"  'scan.startup.mode' = 'earliest-offset',\n" +
				"  'format' = 'json',\n" +
				"  'json.fail-on-missing-field' = 'false',\n" +
				"  'json.ignore-parse-errors' = 'true'\n" +
				")"
		);

		// 3.查询分析-select
		Table table = tabEnv.sqlQuery(
			"SELECT\n" +
				"    after['province_id'] AS province_id,\n" +
				"    count(after['id']) AS order_count,\n" +
				"    sum(CAST(after['total_amount'] AS DOUBLE)) AS order_amount\n" +
				"FROM mysql_cdc_kafka_source\n" +
				"WHERE after IS NOT NULL\n" +
				"GROUP BY after['province_id']"
		);
		//table.execute().print();

		// 4.输出表-OutputTabel
		tabEnv.executeSql(
			"CREATE TABLE print_table (\n" +
				"    province_id STRING,\n" +
				"    order_count BIGINT,\n" +
				"    order_amount DOUBLE\n" +
				") WITH (\n" +
				"    'connector' = 'print'\n" +
				")"
		);

		// 5.插入保存-insert
		tabEnv.createTemporaryView("result_table", table);
		tabEnv.executeSql(
			"INSERT INTO print_table \n" +
				"SELECT province_id, order_count, order_amount FROM result_table"
		);

	}

}
